package com.zx.mes.hyl.producer;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.zx.mes.hyl.entity.CustomerAccount;
import com.zx.mes.hyl.mapper.CustomerAccountMapper;
import lombok.extern.log4j.Log4j2;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.math.BigDecimal;
import java.time.LocalDateTime;
import java.util.Map;

/**
 * 客户付款要处理的业务
 *
 * @author 华云龙
 * @date 2018-12-14
 */
@Log4j2
@Component
public class PayTransactionListener implements TransactionListener {

    @SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection")
    @Autowired
    private CustomerAccountMapper customerAccountMapper;

    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        log.debug("执行本地事务,发送的消息msg=[" + msg + "],所带参数arg=[" + arg + "]");
        try {
            /// 将arg转成 CustomerAccount

            Map map = (Map) arg;
            String accountId = (String) map.get("accountId");
            Integer version = (Integer) map.get("version");
            BigDecimal money = (BigDecimal) map.get("money");
            BigDecimal currentBalance = (BigDecimal) map.get("currentBalance");

            /// 做乐观锁去重
            CustomerAccount customerAccount = new CustomerAccount();
            customerAccount.setAccountId(accountId)
                    .setVersion(version + 1)
                    .setUpdateTime(LocalDateTime.now())
                    .setCurrentBalance(currentBalance.subtract(money));
            int update = customerAccountMapper.update(customerAccount, new QueryWrapper<CustomerAccount>()
                    .lambda()
                    .eq(CustomerAccount::getAccountId, accountId)
                    .eq(CustomerAccount::getVersion, version));
            if (update == 1) {
                log.debug("本地事务成功");
                return LocalTransactionState.COMMIT_MESSAGE;
            } else {
                log.error("本地事务失败,消息直接回滚,回滚原因(更新):update=[" + update + "]条数据!");
                return LocalTransactionState.ROLLBACK_MESSAGE;
            }
        } catch (Exception e) {
            e.printStackTrace();
            log.error("本地事务失败:exception=[" + e.getMessage() + "]");
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }

    }

    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        /// 回检不做,本地事务不成功,消息直接回滚
        return null;
    }
}
